package dws

import java.time.{LocalDateTime, ZoneOffset}
import java.time.format.DateTimeFormatter
import java.util

import modes.OrderDetails
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.api.common.eventtime.{Watermark, WatermarkGenerator, WatermarkGeneratorSupplier, WatermarkOutput, WatermarkStrategy}
import org.apache.flink.cep.functions.{PatternProcessFunction, TimedOutPartialMatchHandler}
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object TradeOrdersCheck {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val data: DataStream[OrderDetails] = env.fromElements(
      OrderDetails(9390, 1, "2020-07-28 00:15:11",295),
      OrderDetails(5990, 1, "2020-07-28 00:16:12",165),
      OrderDetails(9390, 2, "2020-07-28 00:18:11",295),
      OrderDetails(5990, 2, "2020-07-28 00:18:12",165),
      OrderDetails(9390, 3, "2020-07-29 08:06:11",295),
      OrderDetails(5990, 4, "2020-07-29 12:21:12",165),
      OrderDetails(8457, 1, "2020-07-30 00:16:15",132),
      OrderDetails(5990, 5, "2020-07-30 18:13:24",165),
      OrderDetails(1001, 1, "2020-10-20 11:05:15",132),
      OrderDetails(1001, 2, "2020-10-20 11:25:15",132),
      OrderDetails(8458, 2, "2020-10-20 11:00:15",132))
    val format: FastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")
    val keyed: KeyedStream[OrderDetails, Int] = data.assignTimestampsAndWatermarks(new WatermarkStrategy[OrderDetails] {
      override def createWatermarkGenerator(context: WatermarkGeneratorSupplier.Context): WatermarkGenerator[OrderDetails] = {
        new WatermarkGenerator[OrderDetails] {
          var watermark = 0L
          val delay: Long = 5000L

          override def onEvent(event: OrderDetails, eventTimestamp: Long, output: WatermarkOutput): Unit = {
            if (eventTimestamp - delay > watermark) {
              watermark = eventTimestamp - delay
            }
          }

          override def onPeriodicEmit(output: WatermarkOutput): Unit = {
            output.emitWatermark(new Watermark(watermark))
          }
        }
      }
    }).assignAscendingTimestamps(x => {
      format.parse(x.date).getTime
    }).keyBy(x => x.orderId)

    val pattern: Pattern[OrderDetails, OrderDetails] = Pattern.begin[OrderDetails]("start")
      .where(x => x.status == 1)
      .followedBy("second")
      .where(x => x.status == 2)
      .within(Time.minutes(15))

    val patterned: PatternStream[OrderDetails] = CEP.pattern(keyed, pattern)
    val outputTag = new OutputTag[OrderDetails]("unmatched")
    val result: DataStream[OrderDetails] = patterned.process(new MyPatternProcessFunction[OrderDetails, OrderDetails](outputTag))

//    result.print()
    result.getSideOutput(outputTag).print()
    env.execute()
  }
  class MyPatternProcessFunction[IN, OUT](outputTag: OutputTag[OrderDetails]) extends PatternProcessFunction[IN, OUT] with TimedOutPartialMatchHandler[IN] {
    override def processMatch(map: util.Map[String, util.List[IN]], context: PatternProcessFunction.Context, collector: Collector[OUT]): Unit = {
      println("matched:" + map)
      collector.collect(map.get("start").get(0).asInstanceOf[OUT])
    }

    override def processTimedOutMatch(map: util.Map[String, util.List[IN]], context: PatternProcessFunction.Context): Unit = {
      println("unmatched:" + map)
      context.output(outputTag, map.get("start").get(0).asInstanceOf[OrderDetails])
    }
  }
}
